#pragma once

#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"

#include <memory>
#include <string>
#include <list>
#include <mutex>
#include <unordered_map>

namespace mq
{
// 我们的消息数据持久化是存放在普通文件中的因为有些消息会很大，不适合存放数据库，其次，我们主要是为了备份，不涉及查询。
// 持久化文件命名:       qname.mqd      qname.mqd.tmp
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<Message>;
    class MessageManpper
    {
    private:
        std::string _qname;    // 队列名称
        std::string _datafile; // 保存消息持久化数据文件
        std::string _tmpfile;  // 垃圾回收时的临时文件
    private:
        bool insert(const std::string &file, MessagePtr &msp)
        {
            // 将消息数据中的有效载体序列化
            std::string body = msp->payload().SerializeAsString();

            FileHelper helper(file);
            size_t fsize = helper.size();
            size_t msg_size = body.size();
            // 往文件写入4字节数据长度
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("往队列文件 %s 写入数据长度失败", file.c_str());
                return false;
            }
            // DLOG("往队列文件 %s 写入数据长度：%d",file.c_str(),msg_size);

            // 往文件写入消息有效载荷
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), msg_size);
            if (ret == false)
            {
                DLOG("往队列文件 %s 写入数据主体失败", file.c_str());
                return false;
            }
            // DLOG("往队列文件 %s 写入数据：%d",file.c_str(),body.c_str());

            // 更改消息数据的偏移量和长度
            msp->set_offset(fsize + sizeof(size_t));
            msp->set_length(msg_size);
            return true;
        }

        bool load(std::list<MessagePtr> &result)
        {
            // 将文件中的有效数据加载出来
            FileHelper data_file_helper(_datafile);
            bool ret;
            size_t msg_size = 0;                     // 四字节长度
            size_t offset = 0;                      // 偏移量
            size_t fsize = data_file_helper.size(); // 文件长度

            // 循环读取文件中的消息
            while (offset < fsize)
            {
                // 读取四字节长度
                ret = data_file_helper.read((char *)&msg_size, offset,sizeof(size_t));
                if (ret == false)
                {
                    DLOG("读取队列文件 %s 数据长度失败", _datafile);
                    return false;
                }
                // 移动偏移量
                offset += sizeof(size_t);

                // 读取有效载体
                std::string msg_body(msg_size, '\0');
                ret = data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取队列文件 %s 消息数据失败", _datafile);
                    return false;
                }

                // 移动偏移量
                offset += msg_size;

                // 构建信息对象
                MessagePtr msp = std::make_shared<Message>();

                // 将消息数据反序列化
                msp->mutable_payload()->ParseFromString(msg_body);

                // 如果是无效消息，则直接处理下一个
                if (msp->payload().valid() == "0")
                {
                    DLOG("加载到无效消息：%s", msp->payload().body().c_str());
                    continue;
                }

                // 有效消息则保存起来
                result.push_back(msp);
            }
            return true;
        }

    public:
        MessageManpper(std::string basedir, const std::string &qname) : _qname(qname)
        {
            // 创建目录
            FileHelper::createDirectory(basedir);
            if (basedir.back() != '/')
                basedir.push_back('/');

            // 拼接文件名
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;

            // 创建文件
            assert(createMsgFile());
        }
        bool createMsgFile()
        {
            //如果文件存在，直接return
            if(FileHelper(_datafile).exists() == true) return true;

            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建队列文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }
        bool removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(MessagePtr &msp)
        {
            insert(_datafile, msp);
        }
        bool remove(MessagePtr &msp)
        {
            // 将消息数据的有效位设置位'0'
            msp->mutable_payload()->set_valid("0");
            // 对msg进行序列化
            std::string body = msp->payload().SerializeAsString();
            if (body.size() != msp->length())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致!");
                return false;
            }

            FileHelper helper(_datafile);
            // 将消息数据覆盖写入到文件位置
            bool ret = helper.write(body.c_str(), msp->offset(), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件 %s 写入数据失败", _datafile);
                return false;
            }
            return true;
        }
        // 恢复数据/垃圾回收
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> result;

            bool ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效数据失败！\n");
                return result;
            }

            // DLOG("加载有效数据结束,数据个数:%d",result.size());

            // 将有效数据写入临时文件
            FileHelper::createFile(_tmpfile);   //必须先创建出临时文件，在datafile中没有数据时，不会进这个循环。会导致文件源文件被删除，tmp文件也没了
            FileHelper tmp_file_helper(_tmpfile);
            size_t offset = 0;
            for (auto &msp : result)
            {
                ret = insert(_tmpfile, msp);
                if (ret == false)
                {
                    DLOG("向临时文件写入消息数据失败！！");
                    return result;
                }
            }
            // DLOG("像临时文件写入数据结束,临时文件大小:%d",tmp_file_helper.size());
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败！");
                return result;
            }

            // 4. 修改临时文件名，为源文件名称
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("修改临时文件名称失败！");
                return result;
            }

            // 5. 返回新的有效数据
            return result;
        }
    };

    class QueueMessage
    {
    private:
        MessageManpper _manpper;    //持久化操作的句柄
        std::string _qname;  // 队列名
        size_t _valid_count; // 有效消息数量
        size_t _total_count; // 消息总数量
        std::mutex _mutex;
        std::list<MessagePtr> _msgs;                               // 带推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息
    private:
        // 当持久化消息总数量大于2000,并且有效消息占比低于总消息的%50，进行垃圾回收
        bool GCCheck()
        {
            if (_total_count >= 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }

            return false;
        }

        void gc()
        {
            if (GCCheck() == false)
                return;
            // 获取有效消息
            std::list<MessagePtr> msgs = _manpper.gc();//这里构造了一个list<MessagePtr>，他和我们的_durable_msgs中的MessagePtr是不同的，所以需要单独更改_durable_msgs中MessagePtr的偏移量
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    // 持久化文件中的消息，在内存中不存在
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理");
                    _msgs.push_back(it->second); // 做法：将该消息插入进待推送链表的末尾
                    _durable_msgs.insert({msg->payload().properties().id(), msg});
                    continue;
                }

                // 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }

            // 3. 更新当前的有效消息数量 & 总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

    public:
        using ptr = std::shared_ptr<QueueMessage>;
        // 创建/打开队列消息文件
        QueueMessage(std::string &basedir, const std::string &qname)
            : _manpper(basedir, qname), _valid_count(0), _total_count(0)
        {
            // 没有在构造函数中恢复信息数据,而是提供了一个recover函数,供MessageManager调用
        }

        // 恢复历史数据
        void recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 获取有效消息
            _msgs = _manpper.gc();

            for (auto &msg : _msgs)
            {
                // 存储进持久化哈希表中
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }

            _valid_count = _total_count = _msgs.size();
        }

        bool insert(const BasicProperties *bp, std::string body, bool queue_is_durable)
        {
            // 构造消息对象
            MessagePtr msp = std::make_shared<Message>();
            msp->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                // 用户设置了属性
                // 可投递模式，需要根据队列的持久化标志判断
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_id(bp->id());
                msp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                msp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            else
            {
                // 用户没有设置属性
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msp->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mutex);
            // 判断信息是否需要持久化
            if (msp->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 设置有效位位字符'1'
                msp->mutable_payload()->set_valid("1");
                // 持久化存储
                bool ret = _manpper.insert(msp);
                if (ret == false)
                {
                    ELOG("信息持久化存储失败, %s", msp->payload().body().c_str());
                    // TODO需要释放掉构造的MessagePtr对象
                    return false;
                }

                // 更新持久化消息数量
                _valid_count += 1;
                _total_count += 1;

                // 存储在持久化消息哈希表
                _durable_msgs.insert({msp->payload().properties().id(), msp});
            }

            // 将信息存储至带推送信息链表
            _msgs.push_back(msp);
            return true;
        }

        //这个删除是在客户端确认之后进行的，从等待确认哈希表中删除
        // 每次删除后，进行一次垃圾回收判断
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                // 要删除的信息不存在
                return true;
            }

            // 判断信息是否被持久化
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 删除持久化信息
                bool ret = _manpper.remove(it->second);
                if (ret == false)
                {
                    ELOG("删除持久化信息失败, %s", it->second->payload().properties().id());
                    return false;
                }

                // 更新持久化信息有效个数
                _valid_count -= 1;

                // 删除持久化消息哈希表中消息
                _durable_msgs.erase(msg_id);

                // 进行垃圾回收
                gc();
            }

            // 删除待确任哈希表中的信息
            _waitack_msgs.erase(msg_id);
            return true;
        }

        // 获取队首消息
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);

            if(_msgs.size() == 0){
                return MessagePtr();
            }

            // 从待推送链表中取出一个消息
            MessagePtr msp = _msgs.front();
            _msgs.pop_front();

            // 并将该消息添加进待确认哈希表
            _waitack_msgs.insert({msp->payload().properties().id(), msp});

            return msp;
        }
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear()
        {
            _manpper.removeMsgFile();
            _valid_count = 0;
            _total_count = 0;
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
        }
    };

    class MessageManager
    {
    private:
        std::string _basedir; //存储队列信息文件的路径，我们的队列信息是存储在文件中的
        std::mutex _mutex;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; //队列名对应队列信息

    public:
    using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string &basedir) : _basedir(basedir) {}
        // 初始化一个队列消息管理,在创建队列的时候调用
        void InitQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr tmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    //DLOG("消息管理句柄 %s 已经存在", qname.c_str());
                    return;
                }

                // 构建一个队列消息内存管理类对象
                QueueMessage::ptr qmp = std::make_shared<QueueMessage>(_basedir, qname);
                tmp = qmp;
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }

            // 恢复历史数据，这个操作是非常耗时的，我们没有放在加锁里。
            tmp->recovery();
        }
        // 销毁一个队列消息管理，在删除队列的时候调用
        void DestoryQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr tmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("消息管理句柄 %s 不存在", qname.c_str());
                    return;
                }

                tmp = it->second;
                _queue_msgs.erase(qname);
            }

            tmp->clear();
        }
        // 新增一条消息
        bool insert(const std::string &qname, BasicProperties *bp, std::string body, bool queue_is_durable)
        {
            QueueMessage::ptr tmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列%s新增消息失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }

                tmp = it->second;
            }

            return tmp->insert(bp, body, queue_is_durable);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr tmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列%s获取队首消息失败:没有找到消息管理句柄!", qname.c_str());
                    return MessagePtr();
                }

                tmp = it->second;
            }

            return tmp->front();
        }
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr tmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列%s确认消息失败:没有找到消息管理句柄!", qname.c_str());
                    return;
                }

                tmp = it->second;
            }

            tmp->remove(msg_id);
            return;
        }

        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &msg : _queue_msgs)
            {
                msg.second->clear();
            }
            _queue_msgs.clear();
        }
    };
}